package com.learn.lb.storm.wordcount;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

/**
 * 处理spout发送过来的数据
 * @author laibo
 * @since 2019/9/18 09:33
 */
public class WcBolt extends BaseRichBolt {

    private Logger logger = LoggerFactory.getLogger(WcBolt.class);

    private Map<String, Integer> dataMap = Collections.synchronizedMap(new HashMap<>(8));

    /**
     * 为bolt提供要执行的环境。执行器将运行此方法来初始化spout
     * @param stormConf 为此bolt提供Storm配置。
     * @param context 提供有关拓扑中的bolt位置，其任务ID，输入和输出信息等的完整信息。
     * @param collector 使我们能够发出处理的元组。
     */
    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {

    }

    /**
     * 收到 spout 发送来的 Word 进行统计
     * @param input 接受到的数据
     */
    @Override
    public void execute(Tuple input) {
        String word = input.getString(0);
        logger.info("接受到的数据：{}", word);
        Integer oldCount = Optional.ofNullable(dataMap.putIfAbsent(word, 0)).orElse(0);
        dataMap.put(word, ++oldCount);
        logger.info("单词{}出现了{}次", word, oldCount);
    }

    /**
     * 如果需要发送到下一个Bolt中处理，在改方法定义发射字段类型
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}
